/**   
 * @{#} Group.java Create on 2013-6-22 下午10:22:35   
 *   
 * Copyright (c) 2012 by Baidu.   
 */
package objtable.algm;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

import objtable.core.GroupTableGenerator;
import objtable.core.IntKeyGen;
import objtable.core.TableCenter;
import objtable.core.TableGenerator;
import objtable.struct.DataType;
import objtable.struct.Row;

import com.google.common.base.Function;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;

/**
 * @author <a href="mailto:zhangbo07@baidu.com">zhangbo07</a>
 * @version 1.0
 */

public class GroupBy implements Algm1Row {
	String[] nfields;
	String[] nsumfields;
	TableGenerator otg;
	GroupTableGenerator ntg;

	public GroupTableGenerator getGroupTableGenerator() {
		return ntg;
	}

	/**
	 * 将传入的数据根据设定的结果进行分组
	 * 
	 * @param oldtablename
	 *            待分组数据表名称
	 * @param groupfieldstr
	 *            分组用的字段，比如：name,age
	 * @param sumfieldstr
	 *            分组后组内求和用的字段，比如：divide
	 * @param newtablename
	 *            因为分组动作会根据传入的groupfieldstr，sumfieldstr构造出新的Row，这里给出的是新Row的名称
	 */
	public GroupBy(String groupfieldstr, String sumfieldstr) {
		nfields = groupfieldstr.split(",");
		nsumfields = sumfieldstr.split(",");
	}

	/**
	 * 生成分组用的方法，其实就是根据groupfield做联合主键
	 * 
	 * @param fields
	 * @return
	 */
	private Function<Row, String> genGroupFunc(final String[] fields) {
		Function<Row, String> func;
		func = new Function<Row, String>() {
			public String apply(Row input) {
				String key = "";
				for (String f : fields) {
					key += (input.get(f) + "^");
				}
				return key;
			}
		};
		return func;
	}

	/**
	 * 计算汇总数据
	 * 
	 * @param field
	 *            需要汇总的字段
	 * @param subs
	 *            需要汇总的数据
	 * @return
	 */
	private Object sum(String field, List<Row> subs) {
		DataType type = getFieldType(field);
		switch (type) {
		case INT:
			return intSum(field, subs);
		case LONG:
			return longSum(field, subs);
		case DOUBLE:
			return doubleSum(field, subs);
		}
		return null;
	}

	private Object doubleSum(String field, List<Row> subs) {
		double result = 0.0;
		for (Row r : subs) {
			result += r.getDouble(field);
		}
		return result;
	}

	private Object longSum(String field, List<Row> subs) {
		long result = 0l;
		for (Row r : subs) {
			result += r.getLong(field);
		}
		return result;
	}

	private Object intSum(String field, List<Row> subs) {
		int result = 0;
		for (Row r : subs) {
			result += r.getInt(field);
		}
		return result;
	}

	private DataType getFieldType(String field) {
		return otg.getFieldType(field);
	}

	public List<? extends Row> execute(List<? extends Row> rows) {
		if (rows.size() == 0) {
			return new ArrayList<Row>();
		} else if (otg == null && rows.size() > 0) {
			otg = TableCenter.getTable(rows.get(0).getTableName());
		}

		if (ntg == null) {
			ntg = new GroupTableGenerator("tmp_" + IntKeyGen.nextInt());
			for (String s : nfields) {
				ntg.addField(otg.getFieldType(s), s);
			}
			for (String s : nsumfields) {
				ntg.addField(otg.getFieldType(s), s);
			}
		}

		@SuppressWarnings("unchecked")
		final Multimap<String, Row> map = (Multimap<String, Row>) Multimaps
				.index(rows, genGroupFunc(nfields));
		Set<String> keySet = map.keySet();
		List<Row> result = new ArrayList<Row>(keySet.size());

		for (String k : keySet) {
			Row nr = ntg.newRow();
			ArrayList<Row> subs = new ArrayList<Row>(map.get(k));
			Row ot = subs.get(0);

			for (String s : nfields) {
				nr.set(s, ot.get(s));
			}
			for (String s : nsumfields) {
				nr.set(s, sum(s, subs));
			}
			nr.set("t_sub", subs);

			result.add(nr);
		}
		return result;
	}

}
